package cn.doitedu.sync;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


/**
 * @Author: deep as the sea
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2022/12/10
 * @Tips: 学大数据，到多易教育
 * @Desc:
 *  视频内容信息，从业务库实时同步到hbase
 *  核心要点： 利用mysql-cdc-connector监听mysql的binlog来获取数据
 *  然后对数据稍作整理，利用hbase-connector写入hbase
 **/
public class S02_SyncJob_VideoInfoTable2Hbase {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/checkpoint");
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 创建业务库的 ums_member 表 的cdc连接器表
        // 挑选一些关心的维度字段：  id,username,phone,status,create_time,gender,province,city,job
        tEnv.executeSql("CREATE TABLE cms_video_source (    " +
                "      id INT,                                      " +
                "      video_type STRING,                           " +
                "      video_album STRING,                          " +
                "      video_author STRING,                          " +
                "      video_timelong BIGINT,                        " +
                "      create_time timestamp(3),                    " +
                "     PRIMARY KEY (id) NOT ENFORCED            " +
                "     ) WITH (                                 " +
                "     'connector' = 'mysql-cdc',               " +
                "     'hostname' = 'doitedu'   ,               " +
                "     'port' = '3306'          ,               " +
                "     'username' = 'root'      ,               " +
                "     'password' = 'root'      ,               " +
                "     'database-name' = 'realtimedw',          " +
                "     'table-name' = 'cms_video'              " +
                ")");


        // 创建hbase连接器sink表
        tEnv.executeSql("CREATE TABLE dim_video_info_hbasesink( " +
                " id INT, " +
                " f ROW<video_type STRING, video_album STRING, video_author STRING,video_timelong BIGINT, create_time TIMESTAMP(3)>, " +
                " PRIMARY KEY (id) NOT ENFORCED " +
                ") WITH (                             " +
                " 'connector' = 'hbase-2.2',          " +
                " 'table-name' = 'dim_video_info',     " +
                " 'zookeeper.quorum' = 'doitedu:2181' " +
                ")");

        // 将数据，写入hbase表
        tEnv.executeSql(/*"insert into dim_video_info_hbasesink " +*/
                "select id,ROW(video_type , video_album , video_author ,video_timelong , create_time) from cms_video_source").print();
    }
}
